package edu.csl.study.spark.basic

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 日志分析：PV UV TOPN
 */
object Core_LogAnalysis {
  /**
   * 获取上下文
   *
   * @return
   */
  def newContext:SparkContext={
    //注意：线程数不能设置为1，必须大于1 .因为必须有一个线程接受数据，其他线程处理数据
    val conf:SparkConf = new SparkConf().setAppName("LogAnalysis").setMaster("local[2]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")
    sc
  }
  val rootDir = System.getProperty("user.dir")+ "\\testFile\\"
  def main(args: Array[String]): Unit = {
      // 1.配置信息 ,没有参数，可以省略括号
      val sc = newContext
      //2.数据输入 :
      val rdd:RDD[String] = sc.textFile(rootDir+"access.log",4)
      /*****************PV****************/
      val count = rdd.count()
      println("PV :"+count)
      /*****************UV****************/
      //取第一条记录
      val rdd_distinct_ip = rdd.map(x => x.split(" ")(0)).distinct();
      //IP去重
      val count_distinct_ip = rdd_distinct_ip.count()
      println("UV :"+count_distinct_ip)
      /*****************TopN****************/
       //长度大于10的行组成的RDD
      val rdd_valid = rdd.filter(x=>x.split(" ").length>10)
      //切分每一行，过滤出丢失的字段数据，获取页面地址
      val rdd_map = rdd_valid.map(_.split(" ")(10)).map((_,1))
      val  rdd_reduce = rdd_map.reduceByKey(_+_)
      //排序 :降序
      val rdd_sort = rdd_reduce.sortBy(t => t._2, false)
      //取出前5条
      val top_5 = rdd_sort.take(5)
      top_5.foreach(println)
      //3.处理过程

      //4.数据输出

      //关闭上下文
      sc.stop()
  }


}
